﻿/*
 * Copyright 2009 - 2010 eScience Institute
 *
 * This file is part of SqlShare.
 *
 * SqlShare is free software: you can redistribute it and/or modify it under the terms 
 * of the GNU General Public License as published by the Free Software Foundation, 
 * either version 3 of the License, or (at your option) any later version.
 *
 * SqlShare is distributed in the hope that it will be useful, but WITHOUT ANY 
 * WARRANTY; without even the implied warranty of MERCHANTABILITY or FITNESS FOR 
 * A PARTICULAR PURPOSE. See the GNU General Public License for more details.
 *
 * You should have received a copy of the GNU General Public License along with 
 * SqlShare. If not, see http://www.gnu.org/licenses/.
*/
using System;
using System.Collections.Generic;
using System.Linq;
using System.Text;
using System.Data;
using System.IO;
using System.Data.SqlClient;
using System.Threading;

namespace RawData
{
    public class SqlBatchFileUploader
    {
        private DataTable _dataTable;
        private int _commitBatchSize;
        private bool _hasHeaders;
        private SqlTableMetadata _tableMeta;
        private StreamReader _reader;   // initialized in loadBulk
        private long _fileSize;         // initialized in loadBUlk
        private bool isComplete;

        /// <summary>
        /// 
        /// </summary>
        /// <param name="tb"></param>
        /// <param name="commitBatchSize"></param>
        /// <param name="hasHeaders">does this stream come with column headers?</param>
        public SqlBatchFileUploader(SqlTableMetadata tb, int commitBatchSize, bool hasHeaders)
        {
            _tableMeta = tb;
            _hasHeaders = hasHeaders;
            _dataTable = new DataTable("[" + tb.TableName + "]");
            _commitBatchSize = commitBatchSize;
            isComplete = false;            
        }

        public string TableName
        {
            get { return _tableMeta.TableName; }
        }

        public int RowsUploaded { get; private set; }

        private void InitDataTableStructure()
        {
            int listSize = _tableMeta.DataColumns.Count;
            for (int i = 0; i < listSize; i++)
            {
                SqlColumnMetadata colMeta = _tableMeta.DataColumns[i];
                DataColumn col = new DataColumn(colMeta.ColumnName, SqlTypeToGenericType(colMeta.KnownDBType));
                col.AllowDBNull = true;
                _dataTable.Columns.Add(col);
            }            
        }

        /// <summary>
        /// Get the percent of this operation
        /// </summary>
        public double PercentComplete
        {
            get
            {
                if (isComplete)
                {
                    return 1;
                }

                if (AsyncException != null)
                {
                    return -1;
                }

                if (_reader != null &&_reader.BaseStream != null && _reader.BaseStream.CanSeek)
                {
                    double val = ((double)_reader.BaseStream.Position / (double)_fileSize);
                    if (val >= 1)
                    {
                        return 0.99d;   // we dont want to report 100% if we are not flagged complete
                    }
                    return val;
                }

                return 0;
            }
        }

        private void CreateTable(SqlConnection conn)
        {
            SqlCommand cmd = new SqlCommand(_tableMeta.SQLCreateTable, conn);
            cmd.ExecuteNonQuery();                       
        }

        /// <summary>
        /// Check to see if this table can be created
        /// </summary>
        /// <param name="conn"></param>
        /// <returns></returns>
        public bool TableExist(SqlConnection conn)
        {
            SqlCommand cmd = new SqlCommand("SELECT COUNT(name) FROM sysobjects WHERE xtype IN ('U','V') AND name = @tblname", conn);
            cmd.Parameters.Add("tblname", SqlDbType.NVarChar).Value = _tableMeta.TableName;
            int count = (int) cmd.ExecuteScalar();
            return count == 1;
        }

        private void _BulkCopyFile(object o)
        {
            object[] oa = (object[])o;
                Stream stream = (Stream)oa[0];
                SqlConnection conn = (SqlConnection)oa[1];
                bool async = (Boolean)oa[2];
            try
            {
                // create the actual table on the SQL backend
                CreateTable(conn);

                // create the internal datatable structure
                InitDataTableStructure();

                // open a reader for this file
                _reader = new StreamReader(stream);
                _reader.BaseStream.Position = 0;
                _fileSize = stream.Length;

                // skip first line if it has column headers
                if (_hasHeaders)
                {
                    _reader.ReadLine();    
                }
                string line = _reader.ReadLine();
                int listSize = _tableMeta.DataColumns.Count;
                int recordCount = 0;
                DataRow row;

                while (line != null)
                {
                    if (recordCount >= _commitBatchSize)
                    {
                        WriteToDatabase(conn);
                        recordCount = 0;
                    }

                    // row to add to table
                    row = _dataTable.NewRow();

                    // fill up data
                    String[] dataSplit = line.Split(new char[] { _tableMeta.ColumnDelimiter }, StringSplitOptions.None);
                    for (int i = 0; i < listSize; i++)
                    {
                        // handle the null value if the value is not provided
                        if (_tableMeta.DataColumns[i].KnownDBType != SqlDbType.VarChar && string.IsNullOrEmpty(dataSplit[i]))
                        {
                            row[i] = DBNull.Value;
                        }
                        else
                        {
                            row[i] = dataSplit[i];
                        }
                    }

                    _dataTable.Rows.Add(row);
                    recordCount++;
                    RowsUploaded++;
                    line = _reader.ReadLine();
                }

                // flush out the remainders
                if (recordCount > 0)
                {
                    WriteToDatabase(conn);
                }

                isComplete = true;
            }
            catch (Exception e)
            {
                if (async)
                {
                    AsyncException = e;
                }
                else
                {
                    throw e;
                }
            }
        }

        /// <summary>
        /// Note: user must manage their own connection closing once this operation is complete
        /// </summary>
        /// <param name="datastream"></param>
        /// <param name="conn"></param>
        /// <param name="async"></param>
        public void BulkCopyFile(Stream datastream, SqlConnection conn, bool async)
        {
            if (async)
            {
                Thread t = new Thread(new ParameterizedThreadStart(_BulkCopyFile));
                t.Start(new object[] { datastream, conn, async });
            }
            else
            {
                _BulkCopyFile(new object[] { datastream, conn, async });
            }
        }

        private void WriteToDatabase(SqlConnection conn)
        {
            SqlBulkCopy bulkCopy = new SqlBulkCopy(conn, 
                SqlBulkCopyOptions.FireTriggers | SqlBulkCopyOptions.CheckConstraints | SqlBulkCopyOptions.TableLock, 
                null);

            bulkCopy.DestinationTableName = _tableMeta.TableName;
            bulkCopy.WriteToServer(_dataTable);

            _dataTable.Clear();            
        }

        public Exception AsyncException { get; private set; }


        /// <summary>
        /// Reverse the database type to C# type
        /// </summary>
        /// <param name="type"></param>
        /// <returns></returns>
        private static Type SqlTypeToGenericType(SqlDbType type)
        {
            switch (type)
            {                
                case SqlDbType.Int:
                    return typeof(int);
                case SqlDbType.Float:
                    return typeof(double);
                case SqlDbType.VarChar:                    
                default:
                    return typeof(string);
            }
        }
    }
}
